package ds_industry_2025.ds.YangJuan_2024.tzgc

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.DoubleType

import java.util.Properties

/*
    根据 Hive 的 dwd 库中相关表或MySQL 中 shtd_store 中相关表（order_detail、sku_info），
计算出与用户 id 为 6708 的用户所购买相同商品种类最多的前 10 位用户（只考虑他
俩购买过多少种相同的商品，不考虑相同的商品买了多少次），将 10 位用户 id 进行
输出，若与多个用户购买的商品种类相同，则输出结果按照用户 id 升序排序，输出格
式如下，将结果截图粘贴至客户端桌面【Release\任务 C 提交结果.docx】中对应的任
务序号下；
结果格式如下：
-------------------相同种类前 10 的 id 结果展示为：-------------------
1,2,901,4,5,21,32,91,14,52
 */
object t1 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("t1")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._

    val conn=new Properties()
    conn.setProperty("user","root")
    conn.setProperty("password","123456")
    conn.setProperty("driver","com.mysql.jdbc.Driver")

    val order = spark.read
      .jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false", "order_info", conn)
    val detail = spark.read
      .jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false", "order_detail", conn)

    //  todo 拿到所有数据
//    val source = order.join(detail, detail("order_id") === order("id"))
//      .select("user_id", "sku_id")
//      .distinct()

    val source = spark.table("tzgc.source")
      .distinct()

    //  todo 拿到6708用户的购买的商品id
    val user_6708_skus = source.filter(col("user_id") === 6708)
      .select("sku_id")
      .map(_(0).toString.toDouble)
      .collect()

    // 判断
    val result = source.filter(col("user_id") !== 6708)
      .withColumn(
        "p",
        when(col("sku_id").cast(DoubleType).isin(user_6708_skus: _*), lit(1.0)).otherwise(lit(0.0))
      )
      .groupBy("user_id")
      .agg(sum("p").as("some"))
      .orderBy(desc("some"))
      .limit(10)
      .select("user_id")
    println("-----------相同种类前10的id结果展示为-----------------")
    println(result.map(_(0).toString.toInt).collect().mkString(","))





    spark.close()

  }

}
